/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */
package com.sys.midware.hbase.hcore.query;

import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.BitComparator;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.NullComparator;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import com.sys.midware.hbase.hcore.Optimize.HbaseMappingPool;
import com.sys.midware.hbase.hcore.Optimize.QueryOptimze;
import com.sys.midware.hbase.hcore.Optimize.TableOptimize;
import com.sys.midware.hbase.hcore.filter.HbaseFilter;
import com.sys.midware.hbase.hcore.filter.HbaseFilterFactory;
import com.sys.midware.hbase.model.HbaseColumn;
import com.sys.midware.hbase.model.HbaseRow;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * 
 * <b>ClassName：</b> QueryFactory <br/>
 * <b>Description：</b> TODO <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月28日 下午2:59:13 <br/>
 * <b>@version: </b> <br/>
 */
public class QueryFactory {
    private static final Logger logger = LoggerFactory.getLogger(QueryFactory.class);

    private HTableInterface table = null;

    private TableOptimize hbaseOptimizeParam = null;

    public QueryFactory(HTableInterface table) {
        this.table = table;
        this.hbaseOptimizeParam = HbaseMappingPool.map_tableOptimization.get(table.getName());
    }

    /**
     * query by rowkey
     * 
     * @param rowKey
     * @return
     * @throws Exception
     */
    public HbaseRow queryByCondition(byte[] rowKey) throws Exception {
        HbaseRow hbaseRow = new HbaseRow();
        try {
            // 根据rowkey查询
            Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
            Get scan = new Get(rowKey);
            if (this.hbaseOptimizeParam != null) {
                optimizeQuery(scan);
            } else {// default
                scan.setCacheBlocks(true);
            }
            Result r = table.get(scan);
            for (Cell cell : r.rawCells()) {
                HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                        new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                        cell.getTimestamp());
                columns.add(e);
            }
            if (columns.isEmpty()) {
                return null;
            }
            hbaseRow.setColumns(columns);// set columns
            return hbaseRow;
        } catch (Exception e) {
            logger.error("", e.getCause());
            e.printStackTrace();
            throw e;
        }
    }
    
    /**
     * query by rowkey
     * 
     * @param rowKey
     * @return
     * @throws Exception
     */
    public HbaseRow queryByCondition(byte[] rowKey,Filter filter) throws Exception {
        HbaseRow hbaseRow = new HbaseRow();
//        List<HbaseRow> list = new ArrayList<HbaseRow>();
        List<Cell> cellList =null;
        try {
            Get get = new Get(rowKey);
            if (this.hbaseOptimizeParam != null) {
                optimizeQuery(get);
            } else {// default
                get.setCacheBlocks(true);
            }
            if(filter!=null){
                get.setFilter(filter);
            }
            
            Result r = table.get(get);
            cellList = r.listCells();
            if(cellList ==null){
                return hbaseRow;
            }
           
            Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
            for(Cell cell  :cellList){
                HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                        new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                        cell.getTimestamp());
                columns.add(e);
            }
            hbaseRow.setColumns(columns);
            return hbaseRow;
        } catch (Exception e) {
            logger.error("", e.getCause());
            throw e;
        }
    }

    /**
     * 单条件按查询，查询多条记录
     * 
     * @param filter
     */
    public List<HbaseRow> queryByCondition(Filter filter) throws Exception {
        List<HbaseRow> list = new ArrayList<HbaseRow>();
        ResultScanner rs = null;
        try {
            Scan s = new Scan();
            if (this.hbaseOptimizeParam != null) {
                optimizeQuery(s);
            }
            s.setFilter(filter);
            rs = table.getScanner(s);
            for (Result r : rs) {
                HbaseRow hbaseRow = new HbaseRow(r.getRow());
                Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
                for (Cell cell : r.rawCells()) {
                    HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                            cell.getTimestamp());
                    columns.add(e);
                }
                hbaseRow.setColumns(columns);// add columns
                list.add(hbaseRow);
            }
        } catch (Exception e) {
            logger.error("", e);
            throw e;
        } finally {
            close(rs);
        }
        return list;
    }

    /**
     * 组合条件查询
     * 
     * @param filters
     */
    public List<HbaseRow> queryByCondition(List<Filter> filters) throws Exception {
        ResultScanner rs = null;
        List<HbaseRow> list = new ArrayList<HbaseRow>();
        try {
            FilterList filterList = new FilterList(filters);
            Scan scan = new Scan();
            if (this.hbaseOptimizeParam != null) {
                optimizeQuery(scan);
            }
            scan.setFilter(filterList);
            rs = table.getScanner(scan);
            for (Result r : rs) {
                HbaseRow hbaseRow = new HbaseRow(r.getRow());
                Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
                for (Cell cell : r.rawCells()) {
                    HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                            cell.getTimestamp());
                    columns.add(e);
                }
                hbaseRow.setColumns(columns);// add columns
                list.add(hbaseRow);
            }
        } catch (Exception e) {
            logger.error("", e);
            throw e;
        } finally {
            close(rs);
        }
        return list;
    }

    /**
     * start rowKey查询
     * 
     * @param queryParam
     */
    public List<HbaseRow> queryByCondition(HbaseQueryParam queryParam) throws Exception {
        ResultScanner rs = null;
        List<HbaseRow> list = null;
        try {
            // init List
            if (queryParam.getLimit() != -1) {
                list = new ArrayList<HbaseRow>(queryParam.getLimit());
            } else {
                list = new ArrayList<HbaseRow>();
            }
            Scan scan = new Scan();

            // set scan
            optimizeQuery(scan, queryParam);

            if (queryParam.getFilter() != null) {// 单filter
                scan.setFilter(parseFilter(queryParam.getFilter()));
            }

            if (queryParam.getFilters() != null) {// 多filter
                List<Filter> filters = new ArrayList<Filter>();
                for (HbaseFilter hbasefilter : queryParam.getFilters()) {
                    filters.add(this.parseFilter(hbasefilter));
                }
                scan.setFilter(new FilterList(filters));
            }

            if (queryParam.getStartRow() != null) {
                scan.setStartRow(queryParam.getStartRow());
            }
            if (queryParam.getStopRow() != null) {
                scan.setStopRow(queryParam.getStopRow());
            }

            rs = table.getScanner(scan);
            int count = 0;
            for (Result r : rs) {
                HbaseRow hbaseRow = new HbaseRow(r.getRow());
                Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
                for (Cell cell : r.rawCells()) {

                    HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                            cell.getTimestamp());
                    columns.add(e);
                }
                hbaseRow.setColumns(columns);// add columns
                list.add(hbaseRow);// add row
                // count
                if (queryParam.getLimit() != -1 && ++count >= queryParam.getLimit()) {
                    break;
                }
            }
        } catch (Exception e) {
            logger.error("", e);

            throw e;
        } finally {
            close(rs);
        }
        return list;
    }

    /**
     * @param rowKeys
     * @return
     * @throws Exception
     */
    public List<HbaseRow> queryByCondition(Set<String> rowKeys) throws Exception {

        try {

            List<Get> gets = new ArrayList<Get>();
            List<HbaseRow> rows = new ArrayList<HbaseRow>();
            for (String rowKey : rowKeys) {
                Get get = new Get(rowKey.getBytes());
                if (this.hbaseOptimizeParam != null) {
                    optimizeQuery(get);
                } else {// default
                    get.setCacheBlocks(true);
                }
                // just for image table
                get.addColumn("info".getBytes(), "imageValue".getBytes());

                gets.add(get);
            }
            Result[] rs = table.get(gets);
            for (Result r : rs) {
                HbaseRow hbaseRow = new HbaseRow(r.getRow());
                Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
                for (Cell cell : r.rawCells()) {
                    HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                            cell.getTimestamp());
                    columns.add(e);
                    // System.out.println(e.toString());
                }
                if (columns.isEmpty()) {
                    continue;
                }
                // System.out.println(columns.toString());
                hbaseRow.setColumns(columns);// set columns
                rows.add(hbaseRow);
            }
            return rows;
        } catch (Exception e) {
            logger.error("", e.getCause());
            e.printStackTrace();
            throw e;
        }
    }

    /**
     * @param queue
     * @throws Exception
     */
    public void queryByCondition4Queue(HbaseQueryParam queryParam) throws Exception {
        ResultScanner rs = null;
        try {
            LinkedBlockingQueue<HbaseRow>  queue=queryParam.getQueue();
            Scan scan = new Scan();
            scan.setCacheBlocks(false);
            scan.setCaching(50000);
            // set scan
            optimizeQuery(scan, queryParam);

            if (queryParam.getFilter() != null) {// 单filter
                scan.setFilter(parseFilter(queryParam.getFilter()));
            }

            if (queryParam.getFilters() != null) {// 多filter
                List<Filter> filters = new ArrayList<Filter>();
                for (HbaseFilter hbasefilter : queryParam.getFilters()) {
                    filters.add(this.parseFilter(hbasefilter));
                }
                scan.setFilter(new FilterList(filters));
            }

            if (queryParam.getStartRow() != null) {
                scan.setStartRow(queryParam.getStartRow());
            }
            if (queryParam.getStopRow() != null) {
                scan.setStopRow(queryParam.getStopRow());
            }

            rs = table.getScanner(scan);
            int count = 0;
            for (Result r : rs) {
                HbaseRow hbaseRow = new HbaseRow(r.getRow());
                Set<HbaseColumn> columns = new LinkedHashSet<HbaseColumn>();
                for (Cell cell : r.rawCells()) {

                    HbaseColumn e = new HbaseColumn(new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)),
                            cell.getTimestamp());
                    columns.add(e);
                }
                hbaseRow.setColumns(columns);// add columns
                
                
                queue.put(hbaseRow);
//                list.add(hbaseRow);// add row
                // count
                if (queryParam.getLimit() != -1 && ++count >= queryParam.getLimit()) {
                    break;
                }
            }
        } catch (Exception e) {
            logger.error("", e);

            throw e;
        } finally {
            close(rs);
        }
    }

    private byte[] getValue(Object v) {
        if (v instanceof String)
            return ((String) v).getBytes();
        else if (v instanceof Integer)
            return Bytes.toBytes(((Integer) v));
        else if (v instanceof Long)
            return Bytes.toBytes(((Long) v));
        else if (v instanceof Float)
            return Bytes.toBytes(((Float) v));
        else if (v instanceof Double)
            return Bytes.toBytes(((Double) v));
        else if (v instanceof Boolean)
            return Bytes.toBytes(((Boolean) v));
        else if (v instanceof Short)
            return Bytes.toBytes(((Short) v));
        else if (v instanceof BigDecimal)
            return Bytes.toBytes(((BigDecimal) v));
        else
            return Bytes.toBytes((ByteBuffer) v);
    }

    private Filter parseFilter(HbaseFilter hbaseFilter) {
        Filter filter = null;
        if (hbaseFilter instanceof HbaseFilterFactory.SingleColumnValueFilter) {
            byte[] value = getValue(hbaseFilter.getValue());
            filter = new org.apache.hadoop.hbase.filter.SingleColumnValueFilter(hbaseFilter.getFamily().getBytes(),
                    hbaseFilter.getQualifier().getBytes(), HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()),
                    value);
        } else if (hbaseFilter instanceof HbaseFilterFactory.RowFilter) {
            byte[] value = getValue(hbaseFilter.getValue());
            if (hbaseFilter.getComparator() == HbaseFilterFactory.Comparator.BINARY) {
                filter = new org.apache.hadoop.hbase.filter.RowFilter(
                        HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()), new BinaryComparator(value));
            } else if (hbaseFilter.getComparator() == HbaseFilterFactory.Comparator.REGEX_STRING) {
                filter = new org.apache.hadoop.hbase.filter.RowFilter(
                        HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()),
                        new RegexStringComparator((String) hbaseFilter.getValue()));
            } else if (hbaseFilter.getComparator() == HbaseFilterFactory.Comparator.SUB_STRING) {
                filter = new org.apache.hadoop.hbase.filter.RowFilter(
                        HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()),
                        new SubstringComparator((String) hbaseFilter.getValue()));
            } else if (hbaseFilter.getComparator() == HbaseFilterFactory.Comparator.BINARY_PREFIX) {
                filter = new org.apache.hadoop.hbase.filter.RowFilter(
                        HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()), new BinaryPrefixComparator(value));
            } else if (hbaseFilter.getComparator() == HbaseFilterFactory.Comparator.BIT) {
                filter = new org.apache.hadoop.hbase.filter.RowFilter(
                        HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()), new BitComparator(value, null));
            } else { // if(hbaseFilter.getComparator()==HbaseFilterFactory.Comparator.NULL)
                filter = new org.apache.hadoop.hbase.filter.RowFilter(
                        HbaseMappingPool.compareOpMap.get(hbaseFilter.getOp()), new NullComparator());
            }
        }

        return filter;
    }

    /**
     * 设置优化参数
     * 
     * @param scan
     */
    private void optimizeQuery(Get scan) {
        if (hbaseOptimizeParam.getQueryOptimze() == null) {
            hbaseOptimizeParam.setQueryOptimze(new QueryOptimze());
            return;
        }
        if (hbaseOptimizeParam.getQueryOptimze().getFamilyReturn() != null) {
            scan.addFamily(hbaseOptimizeParam.getQueryOptimze().getFamilyReturn());
        }
        if (hbaseOptimizeParam.getQueryOptimze().getQualifierReturn() != null
                && hbaseOptimizeParam.getQueryOptimze().getFamilyReturn() != null) {
            scan.addColumn(hbaseOptimizeParam.getQueryOptimze().getFamilyReturn(),
                    hbaseOptimizeParam.getQueryOptimze().getQualifierReturn());
        }
        scan.setCacheBlocks(hbaseOptimizeParam.getQueryOptimze().isEnableBlockCache());
    }

    /**
     * 设置优化参数
     * 
     * @param scan
     */
    private void optimizeQuery(Scan scan) {
        if (hbaseOptimizeParam.getQueryOptimze() == null) {
            hbaseOptimizeParam.setQueryOptimze(new QueryOptimze());
            return;
        }
        if (hbaseOptimizeParam.getQueryOptimze().getBatch() != -1) {
            scan.setBatch(hbaseOptimizeParam.getQueryOptimze().getBatch());
        }
        if (hbaseOptimizeParam.getQueryOptimze().getFamilyReturn() != null) {
            scan.addFamily(hbaseOptimizeParam.getQueryOptimze().getFamilyReturn());
        }
        if (hbaseOptimizeParam.getQueryOptimze().getQualifierReturn() != null
                && hbaseOptimizeParam.getQueryOptimze().getFamilyReturn() != null) {
            scan.addColumn(hbaseOptimizeParam.getQueryOptimze().getFamilyReturn(),
                    hbaseOptimizeParam.getQueryOptimze().getQualifierReturn());
        }
        if (hbaseOptimizeParam.getQueryOptimze().getNumScannerCaching() != -1) {
            scan.setCaching(hbaseOptimizeParam.getQueryOptimze().getNumScannerCaching());
        }
        if (hbaseOptimizeParam.getQueryOptimze().getMaxResultSize() != -1) {
            scan.setMaxResultSize(hbaseOptimizeParam.getQueryOptimze().getMaxResultSize());
        }

        scan.setCacheBlocks(hbaseOptimizeParam.getQueryOptimze().isEnableBlockCache());
    }

    /**
     * 
     * @param scan
     */
    private void optimizeQuery(Scan scan, HbaseQueryParam queryParam) {
        if (queryParam.getQueryOptimze() == null) {
            queryParam.setQueryOptimze(new QueryOptimze());
            return;
        }
        if (queryParam.getQueryOptimze().getBatch() != -1) {
            scan.setBatch(queryParam.getQueryOptimze().getBatch());
        }
        if (queryParam.getQueryOptimze().getFamilyReturn() != null) {
            scan.addFamily(queryParam.getQueryOptimze().getFamilyReturn());
        }
        if (queryParam.getQueryOptimze().getQualifierReturn() != null
                && queryParam.getQueryOptimze().getFamilyReturn() != null) {
            scan.addColumn(queryParam.getQueryOptimze().getFamilyReturn(),
                    queryParam.getQueryOptimze().getQualifierReturn());
        }
        if (queryParam.getQueryOptimze().getNumScannerCaching() != -1) {
            scan.setCaching(queryParam.getQueryOptimze().getNumScannerCaching());
        }
        if (queryParam.getQueryOptimze().getMaxResultSize() != -1) {
            scan.setMaxResultSize(queryParam.getQueryOptimze().getMaxResultSize());
        }
        scan.setCacheBlocks(queryParam.getQueryOptimze().isEnableBlockCache());
    }

    private void close(ResultScanner rs) throws Exception {
        if (rs != null) {
            rs.close();
        }
    }
}// end of QueryFactory
